package com.relytosoft.mizar.analyze.algorithm.face.bitmain.service.impl;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.mizar.analyze.algorithm.base.service.UploadService;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlDataReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlFaceControlReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlStatusReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlStructureControlReq;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.FACE_CONTROL_RAW_DATA_TOPIC;
import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.FACE_RAW_DATA_TOPIC;

/**
 * UploadServiceImpl
 *
 * @Author SunLi
 * @Date: 2022/5/16 13:59
 * @Version 1.0
 */
@Service
public class UploadServiceImpl implements UploadService<BtdlDataReq, BtdlStatusReq, BtdlFaceControlReq, BtdlStructureControlReq> {

    @Resource
    private KafkaProducer kafkaProducer;

    @Override
    public Boolean receiveData(BtdlDataReq dataReq) {
        kafkaProducer.send(FACE_RAW_DATA_TOPIC, JSON.toJSONString(dataReq), true);
        return true;
    }

    @Override
    public Boolean receiveStatus(BtdlStatusReq statusReq) {
        Map<String, Object> info = new HashMap<>();
        info.put("taskId", statusReq.getTaskId());
        info.put("code", statusReq.getCode());
        info.put("msg", statusReq.getMsg());
        KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
        kafkaOpMsg.setData(JSON.toJSONString(info));
        kafkaProducer.send(MediaMqConstants.FACE_ANALYZE_TASK_STATUS_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
        return true;
    }

    @Override
    public Boolean receiveFaceControlData(List<BtdlFaceControlReq> controlReqs) {
        kafkaProducer.send(FACE_CONTROL_RAW_DATA_TOPIC, JSON.toJSONString(controlReqs), true);
        return true;
    }

}
